# -*- coding:utf-8 -*- #

# ----------------------------------------------------------
# Name:        ProcessPoolExecutor_asyncio
# Description: 使用多进程和异步io结合
# Authon:      syl
# Date:        2020-12-03
# ----------------------------------------------------------


import asyncio
import aiohttp
from concurrent.futures import ProcessPoolExecutor, as_completed ,ThreadPoolExecutor
import time


async def post_http():
    # 示例
    url = ''
    data = ''
    async  with aiohttp.ClientSession() as session:
        async with session.post(url=url, data=data, headers={}, timeout=60) as resp:
            r_json = await resp.json()
            return r_json


async def t_handler(data, t_flag, p_flag, semaphore):
    async with semaphore:
        for d in data:
            print(f'pid:{p_flag} tid:{t_flag} data:{d}')
            await asyncio.sleep(1)  # 处理费时的io操作，比如httprequest
    return


def p_handler(datas, p_flag):
    # 线程并发数需要有限制  linux打开文件最大默认为1024 win为509 待确认
    ts = time.time()
    num = 10  # 最大并发数
    count = len(datas)
    block = int(count / num) + 1
    tar_datas = [datas[i * block: (i + 1) * block if (i + 1) * block < count else count] for i in range(num)]
    semaphore = asyncio.Semaphore(num)
    tasks = [t_handler(d, i, p_flag, semaphore) for i, d in enumerate(tar_datas)]

    loop = asyncio.get_event_loop()  # 基于当前线程 ，故在多线程中无法使用 只能在多进程中使用
    loop.run_until_complete(asyncio.wait(tasks))

    loop.close()

    return f'\033[0;32mprocess {p_flag} :cost {time.time() - ts}\033[0m'




if __name__ == '__main__':
    ts = time.time()
    datas = [i for i in range(1000)]
    # datas = [datas[i * 100:(i + 1) * 100] for i in range(10)]  # 每个进程要处理的数据

    # 启动异步io 主线程调用 event_loop 在当前线程下启动异步io 通过并发来实现多线程的效果
    res = p_handler(datas,1)
    print(res)

    # p_num = 10
    # block_len = 100
    #
    # datas = [datas[i * 100:(i + 1) * 100] for i in range(p_num)]  # 每个进程要处理的数据
    # # ProcessPoolExecutor 可能与运行环境有关 官方的 with as 会主动释放线程 导致主线程退出时找不到进程池内进程已经被释放 导致Error in atexit._run_exitfuncs异常
    # executor = ProcessPoolExecutor(p_num)
    # futures = [executor.submit(p_handler, d, p_flag) for p_flag, d in enumerate(datas)]
    # for f in as_completed(futures):
    #     if f.done():
    #             res = f.result()
    #             print(res)

    print(f'Exit!! cost:{time.time() - ts}')
